package com.camemax.controller;

import com.camemax.pojo.ApacheLogEvent;
import com.camemax.pojo.UrlViewCount;
import com.camemax.utils.StreamEnvUtils;
import org.apache.commons.compress.utils.Lists;
import org.apache.flink.api.common.eventtime.SerializableTimestampAssigner;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.functions.AggregateFunction;
import org.apache.flink.api.common.state.ListState;
import org.apache.flink.api.common.state.ListStateDescriptor;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.WindowedStream;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.KeyedProcessFunction;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.SlidingEventTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

import java.net.URL;
import java.sql.Timestamp;
import java.text.SimpleDateFormat;
import java.time.Duration;
import java.util.ArrayList;

/*
* 通过Apache日志文件计算网站访问量 —— uv
* */
public class ApacheLogController {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment streamEnv = StreamEnvUtils.getStreamEnv(1);

        URL resource = ApacheLogEvent.class.getResource("/apache.log");
        DataStream<String> fileDataStream =streamEnv.readTextFile(resource.getPath());


        DataStream<ApacheLogEvent> watermarkDataStream = fileDataStream.map(line -> {
            String[] fields = line.split(" ");
            long time = new SimpleDateFormat("dd/MM/yy:HH:mm:ss").parse(fields[3]).getTime();
            return new ApacheLogEvent(fields[0], fields[1], time, fields[5], fields[6]);
        }).assignTimestampsAndWatermarks(
                WatermarkStrategy.<ApacheLogEvent>forBoundedOutOfOrderness(Duration.ofMinutes(1))
                        .withTimestampAssigner(
                                (SerializableTimestampAssigner<ApacheLogEvent>)
                                        (element, recordTimestamp) -> element.getTimestamp() ));

        WindowedStream<ApacheLogEvent, String, TimeWindow> windowedDataStream = watermarkDataStream.filter(line -> "GET".equals(line.getRequestType()))
                .keyBy(ApacheLogEvent::getUrl) // 按照URL分组
                .window(SlidingEventTimeWindows.of(Time.minutes(10), Time.seconds(5)));

        DataStream<UrlViewCount> aggregateDataStream = windowedDataStream.aggregate(new AggregateFunction<ApacheLogEvent, Long, Long>() {
            @Override
            public Long createAccumulator() {
                return 0L;
            }

            @Override
            public Long add(ApacheLogEvent value, Long accumulator) {
                return accumulator + 1;
            }

            @Override
            public Long getResult(Long accumulator) {
                return accumulator;
            }

            @Override
            public Long merge(Long a, Long b) {
                return a + b;
            }
        }, new WindowFunction<Long, UrlViewCount, String, TimeWindow>(){

            @Override
            public void apply(String s, TimeWindow window, Iterable<Long> input, Collector<UrlViewCount> out) throws Exception {
                out.collect(new UrlViewCount(s, window.getEnd(), input.iterator().next()));
            }
        } );

        aggregateDataStream.keyBy(UrlViewCount::getTimestamp)
                .process(new KeyedProcessFunction<Long, UrlViewCount, String>() {

            ListState<UrlViewCount> userViewCountListState;

            @Override
            public void open(Configuration parameters) {
                userViewCountListState = getRuntimeContext().getListState(
                        new ListStateDescriptor<>("user-view-count", UrlViewCount.class)
                );
            }

            @Override
            public void processElement(UrlViewCount value, Context ctx, Collector<String> out) throws Exception {
                ctx.timerService().registerEventTimeTimer(value.getTimestamp() + 1 );
                userViewCountListState.add(value);
            }

            @Override
            public void onTimer(long timestamp, OnTimerContext ctx, Collector<String> out) throws Exception {
                ArrayList<UrlViewCount> viewCounts = Lists.newArrayList(userViewCountListState.get().iterator());
                viewCounts.sort((urlViewCount1, urlViewCount2) -> urlViewCount2.getViewCount().intValue() - urlViewCount1.getViewCount().intValue());
                StringBuilder result = new StringBuilder();
                result.append(" ==================== \n");
                result.append(" 窗口结束时间: ").append(new Timestamp(timestamp - 1)).append("\n");
                for ( int i = 0 ; i < Math.min(viewCounts.size(),3) ; i ++){

                    UrlViewCount currentViewCount = viewCounts.get(i);
                    result.append(" Top.").append(i+1).append("\t").append("Url: ").append(currentViewCount.getUrl());
                    result.append(" View-Count: ").append(currentViewCount.getViewCount()).append("\n");
                }
                result.append(" ==================== \n\n");
                out.collect(result.toString());
            }
        }).print();

        streamEnv.execute();
    }
}
